package com.xiaonei.storm.adapter;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import Ice.Communicator;
import Ice.InitializationData;
import Ice.ObjectAdapter;
import Ice.ObjectPrx;
import IceStorm.AlreadySubscribed;
import IceStorm.NoSuchTopic;
import IceStorm.TopicExists;
import IceStorm.TopicManagerPrx;
import IceStorm.TopicManagerPrxHelper;
import IceStorm.TopicPrx;

import com.xiaonei.storm.ObserverPrx;
import com.xiaonei.storm.ObserverPrxHelper;

public class IceStormAdapter extends TimerTask {

    /**
     * 主题管理器
     */

    private Log logger = LogFactory.getLog(IceStormAdapter.class);

    private TopicManagerPrx manager;

    private ObjectAdapter adapter;

    /**
     * 订阅主题的map
     */
    private Map<String, TopicPrx> topicPrxMap = new HashMap<String, TopicPrx>();

    /**
     * 发布者的map
     */
    private Map<String, ObserverPrx> observerPrxMap = new HashMap<String, ObserverPrx>();

    /**
     * 接收者的缓存，方便重新订阅
     */
    private Map<String, ArrayList<IceStormListener>> listeners = new HashMap<String, ArrayList<IceStormListener>>();

    private static IceStormAdapter instance;

    private ThreadPoolExecutor pool;

    public ThreadPoolExecutor getThreadPoolExecutor() {
        return pool;
    }

    public static IceStormAdapter getInstance() {
        if (instance == null) {
            synchronized (IceStormAdapter.class) {
                if (instance == null) {
                    instance = new IceStormAdapter();
                }
            }
        }
        return instance;
    }

    private IceStormAdapter() {

        pool = new ThreadPoolExecutor(1, 5, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(
                1000), new ThreadPoolExecutor.AbortPolicy());

        Communicator communicator = Ice.Util.initialize(initData());

        //topicmanager

        ObjectPrx objectPrx = communicator.stringToProxy("AsIceStorm/TopicManager");
        objectPrx = objectPrx.ice_timeout(300);
        manager = TopicManagerPrxHelper.uncheckedCast(objectPrx);

        // objectadapter
        String uuid = UUID.randomUUID().toString();
        communicator.getProperties().setProperty(uuid + ".Endpoints", "tcp");
        communicator.getProperties().setProperty(uuid + ".ThreadPool.Size", "1");
        communicator.getProperties().setProperty(uuid + ".ThreadPool.SizeMax", "1000");
        communicator.getProperties().setProperty(uuid + ".ThreadPool.StackSize", "131072");
        adapter = communicator.createObjectAdapter(uuid);
        adapter.activate();
        // 没半个小时重新订阅，是否需要呢
        new Timer().schedule(this, 1 * 60 * 1000, 1 * 60 * 1000);
    }

    /**
     * 
     * @Title: subscribe
     * @Description: 订阅这个主题
     * @param @param name
     * @param @param listener
     * @return void
     * @throws
     */
    public void subscribe(String name, IceStormListener listener) {
        if (null == name) {
            throw new NullPointerException("the name is null");
        }
        TopicPrx topicPrx = getTopicPrx(name);

        synchronized (this) {
            ArrayList<IceStormListener> list = listeners.get(name);
            if (list == null) {
                list = new ArrayList<IceStormListener>();
                listeners.put(name, list);
            }
            if (!list.contains(listener)) {
                listener.register(adapter);
                list.add(listener);
            }
        }
        try {
            try {
                topicPrx.subscribeAndGetPublisher(null, listener.getProxy());
                logger.info("订阅成功！" + listener);

            } catch (AlreadySubscribed e) {
                logger.error("topic has already Subscribed", e);
            }
        } catch (Exception e) {
            logger.error("订阅失败！", e);
        }
    }

    /**
     * 
     * @Title: unsubscribe
     * @Description: 取消订阅这个主题
     * @param @param name
     * @param @param listener
     * @return void
     * @throws
     */
    public void unsubscribe(String name, IceStormListener listener) {
        boolean found = false;
        synchronized (this) {
            ArrayList<IceStormListener> list = listeners.get(name);
            if (list != null && list.contains(listener)) {
                found = true;
                list.remove(listener);
            }
        }
        if (found) {
            if (topicPrxMap.containsKey(name)) {
                TopicPrx topicPrx = topicPrxMap.get(name);
                topicPrx.unsubscribe(listener.deregister(adapter));
            }
        }
    }

    /**
     * 
     * @Title: notify
     * @Description: 发布消息
     * @param @param name
     * @param @param props
     * @return void
     * @throws
     */
    public void notify(String name, String message) {
        ObserverPrx observerPrx = getObserverPublisherPrx(name);
        observerPrx.update(message);
    }

    /**
     * 
     * @Title: getObserverPrx
     * @Description: 获取发布者
     * @param @param name
     * @param @return
     * @return ObserverPrx
     * @throws
     */
    private ObserverPrx getObserverPublisherPrx(String name) {
        if (!observerPrxMap.containsKey(name)) {
            TopicPrx topicPrx = getTopicPrx(name);
            ObjectPrx pub = topicPrx.getPublisher().ice_oneway();
            ObserverPrx observerPrx = ObserverPrxHelper.uncheckedCast(pub);
            observerPrxMap.put(name, observerPrx);
        }
        return observerPrxMap.get(name);
    }

    /**
     * 
     * @Title: getTopicPrx
     * @Description: 获取主题
     * @param @param name
     * @param @return
     * @return TopicPrx
     * @throws
     */
    private TopicPrx getTopicPrx(String name) {
        if (!topicPrxMap.containsKey(name)) {
            TopicPrx topicPrx = null;
            try {
                topicPrx = manager.retrieve(name);
            } catch (NoSuchTopic ex) {
                logger.error("订阅的主题:" + name + "不存在", ex);

                try {
                    topicPrx = manager.create(name);
                } catch (TopicExists e) {
                    logger.error("订阅的主题:" + name + "已经存在", ex);
                }
            }
            topicPrxMap.put(name, topicPrx);
        }
        return topicPrxMap.get(name);
    }

    /**
     * 1分钟重新订阅一次
     */
    public void run() {
        for (Entry<String, ArrayList<IceStormListener>> entry : listeners.entrySet()) {
            for (IceStormListener listener : entry.getValue()) {
                try {
                    logger.info("重新订阅: " + listener);
                    subscribe(entry.getKey(), listener);
                } catch (Exception e) {
                    logger.error("重新订阅出现问题", e);
                }
            }
        }
    }

    /**
     * 
     * @Title: initData
     * @Description: 初始化ICE参数
     * @param @return
     * @return InitializationData
     * @throws
     */
    public static InitializationData initData() {
        Ice.Properties properties = Ice.Util.createProperties();
        properties.setProperty("Ice.Override.Timeout", "500");
        properties.setProperty("Ice.Override.ConnectTimeout", "300");
        properties.setProperty("IceGrid.InstanceName", "XiaoNeiAntiSpam");
        properties.setProperty("Ice.ThreadPool.Client.StackSize", "65535");
        properties.setProperty("Ice.ThreadPool.Client.StackSize", "65535");
        properties.setProperty("Ice.Default.Locator",
                "XiaoNeiAntiSpam/Locator:default -h  10.3.16.98 -p 4061");
        Ice.InitializationData initData = new InitializationData();
        initData.properties = properties;
        return initData;
    }
}
